package com.zz.study.sharding.component;

import com.alibaba.fastjson.JSON;
import com.zz.study.sharding.util.DateUtil;
import com.zz.study.sharding.util.JsonUtil;
import com.zz.study.sharding.util.UUIDUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.ByteArrayEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.nio.charset.StandardCharsets;
import java.sql.Timestamp;
import java.util.List;
import java.util.Map;

/**
 * created by xjx on  2024-01-22 15:31
 */
@Component
@Slf4j
public class StarRocksStreamWriter {

    @Value("${bydata.feip}")
    private String feIp;

    @Value("${bydata.database}")
    private String databaseName;

    @Value("${bydata.userName}")
    private String userName;

    @Value("${bydata.password}")
    private String password;




    public void write(String tableName, String flushData, String columns) {
        String dsLoadUrl = feIp +":8030";
        try {
            doStreamLoad(tableName, dsLoadUrl, databaseName, userName, password,
                    flushData.getBytes(), columns);
        } catch (Exception e) {
            e.printStackTrace();
            log.error("半云数据库入库异常====="+e.getMessage());
        }
    }


    public void importBeanMapList2Doris(List<Map<String,Object>> tempList, String targetTableName,String columns){
        for(Map<String,Object> bean:tempList){
            if(bean.get("RecordTime") instanceof Timestamp){
                bean.put("RecordTime", DateUtil.getTime(DateUtil.timestampToDate((Timestamp) bean.get("RecordTime"))));
                bean.put("ArchiveTime", DateUtil.getTime(DateUtil.timestampToDate((Timestamp) bean.get("ArchiveTime"))));
            }
        }
        //String columns2 = "id,RadarType,Speed,RecordTime,VesselType,BeidouType,Heading,AISType,SourceNames,Lon,AISName,TarID,Uid,MMSI,Length,State,ArchiveTime,TerminalID,Course,Lat";
        try{
            String flushData = JsonUtil.serialize(tempList);
            if(StringUtils.isNotEmpty(flushData)){

                streamLoad(flushData,columns,targetTableName);

            }
            Thread.sleep(1000);
        }catch (Exception e){

            e.printStackTrace();
            log.error(e.getMessage());
        }
        System.out.println("import success!!!!!!");
    }


    public void streamLoad(String flushData,String columns2,String targetTableName){
        String dsLoadUrl = feIp +":8030";
        try {
            log.error("导入数据大小"+flushData.getBytes().length+"bytes");
            doStreamLoad(targetTableName, dsLoadUrl, databaseName, userName, password,
                    flushData.getBytes(), columns2);
        } catch (Exception e) {
            try {
                log.error("sleep 30s=====因为半云数据库入库异常报错=="+e.getMessage());
                Thread.sleep(30000);
            } catch (Exception interruptedException) {
                interruptedException.printStackTrace();
            }
            e.printStackTrace();
            log.error("半云数据库入库异常====="+e.getMessage());
        }
    }



    public static void doStreamLoad(String tableName, String dsLoadUrl, String database, String username, String password,
                                    byte[] flushData, String columns) throws IOException {
        String label = tableName + "_" + UUIDUtil.create();
        String loadUrl = new StringBuilder("http://")
                .append(dsLoadUrl)
                .append("/api/")
                .append(database)
                .append("/")
                .append(tableName)
                .append("/_stream_load")
                .toString();
        log.debug(String.format("Start to join batch data: bytes[%d] label[%s].", flushData.length, label));

        Map<String, Object> loadResult = doHttpPut(username, password, loadUrl, label, flushData, columns);
        final String keyStatus = "Status";
        if (null == loadResult || !loadResult.containsKey(keyStatus)) {
            throw new IOException("Unable to flush data to Doris: unknown result status.");
        }
        log.debug(new StringBuilder("StreamLoad response:\n").append(JSON.toJSONString(loadResult)).toString());
        if ("Fail".equals(loadResult.get(keyStatus))) {
            throw new IOException(
                    new StringBuilder("Failed to flush data to Doris.\n").append(JSON.toJSONString(loadResult)).toString()
            );
        }
    }


    private static Map<String, Object> doHttpPut(String username, String password, String loadUrl, String label,
                                                 byte[] data, String columns) throws IOException {
        log.info(String.format("Executing stream load to: '%s', size: '%s'", loadUrl, data.length));
        final HttpClientBuilder httpClientBuilder = HttpClients.custom()
                .setRedirectStrategy(new DefaultRedirectStrategy() {
                    @Override
                    protected boolean isRedirectable(String method) {
                        return true;
                    }
                });
        try (CloseableHttpClient httpclient = httpClientBuilder.build()) {
            HttpPut httpPut = new HttpPut(loadUrl);
            httpPut.setHeader("Expect", "100-continue");
            httpPut.setHeader("format", "json");
            httpPut.setHeader("label", label);
            httpPut.setHeader("columns", columns);
            httpPut.setHeader("Content-Type", "application/x-www-form-urlencoded");
            httpPut.setHeader("strip_outer_array", "true");
            httpPut.setHeader("Authorization", getBasicAuthHeader(username, password));
            httpPut.setEntity(new ByteArrayEntity(data));
            httpPut.setConfig(RequestConfig.custom().setRedirectsEnabled(true).build());
            try (CloseableHttpResponse resp = httpclient.execute(httpPut)) {
                HttpEntity respEntity = getHttpEntity(resp);
                if (respEntity == null)
                    return null;
                return (Map<String, Object>) JSON.parse(EntityUtils.toString(respEntity));
            }
        }
    }


    private static String getBasicAuthHeader(String username, String password) {
        String auth = username + ":" + password;
        byte[] encodedAuth = org.apache.commons.codec.binary.Base64.encodeBase64(auth.getBytes(StandardCharsets.UTF_8));
        return new StringBuilder("Basic ").append(new String(encodedAuth)).toString();
    }

    private static HttpEntity getHttpEntity(CloseableHttpResponse resp) {
        int code = resp.getStatusLine().getStatusCode();
        if (200 != code) {
            log.warn("Request failed with code:{}", code);
            return null;
        }
        HttpEntity respEntity = resp.getEntity();
        if (null == respEntity) {
            log.warn("Request failed with empty response.");
            return null;
        }
        return respEntity;
    }
}
